package com.ionehe.group.common.async.mq;

import cn.hutool.json.JSONUtil;
import com.google.common.base.Throwables;
import com.ionehe.group.common.notify.message.factory.Factory;
import com.ionehe.group.common.notify.message.producer.Producer;

import javax.annotation.PostConstruct;

import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.SendStatus;
import org.apache.rocketmq.common.message.Message;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;


/**
 * Copyright (c) 2021 ionehe.com
 * Date: 2021/4/21
 * Time: 1:22 下午
 *
 * @author 2021年 <a href="mailto:a@ionehe.com">秀</a>
 * 【异步消息MQ生产者】
 */
@Component
@Slf4j
public class AsyncMQProducer {

    private static Producer producer;

    @Autowired
    Factory factory;

    @Value("${notify.async.topic-id:-1}")
    private String topic;

    @Value("${notify.async.delay-topic-id:-1}")
    private String delayTopic;


    @PostConstruct
    public void init() {
        try {
            if ("-1".equals(this.topic) && "-1".equals(this.delayTopic)) {
                log.warn("async[]AsyncMQProducer[] topic invalid");

                return;
            }
            if ("-1".equals(this.delayTopic)) {
                log.info("async[]AsyncMQProducer[] delay topic invalid");
                this.delayTopic = this.topic;
            }

            producer = this.factory.createProducer();

            if (producer.isStarted()) {
                log.info("async[]AsyncMQProducer[] already start!");
                return;
            }
            producer.start();
            log.info("async[]AsyncMQProducer[] start success[]topic:{}", this.topic);
        } catch (Exception e) {
            log.error("async[]AsyncMQProducer[]init {}", Throwables.getStackTraceAsString(e));
        }
    }


    @Override
    protected void finalize() throws Throwable {
        super.finalize();


        if (null == producer) {
            return;
        }
        producer.shutdown();
    }


    public SendResult send(String key, byte[] message, String tag) {
        if (producer == null) {
            log.error("async[]AsyncMQProducer[] producer is null");
            return new SendResult(SendStatus.SLAVE_NOT_AVAILABLE, null, null, null, 0L);
        }
        try {
            SendResult result = producer.send(new Message(this.topic, tag, key, message));
            log.info("async[]AsyncMQProducer[] send key:{} tag:{} result:{}", new Object[]{key, tag, result});
            return result;
        } catch (Exception e) {
            log.error("async[]AsyncMQProducer[] send key {} ex {}", key, Throwables.getStackTraceAsString(e));
            return new SendResult(SendStatus.SLAVE_NOT_AVAILABLE, null, null, null, 0L);
        }
    }


    public SendResult send(String key, byte[] msg, String tag, int delayTimeLevel) {
        if (producer == null) {
            return new SendResult(SendStatus.SLAVE_NOT_AVAILABLE, null, null, null, 0L);
        }
        try {
            Message message = new Message(this.delayTopic, tag, key, msg);
            message.setDelayTimeLevel(delayTimeLevel);
            SendResult result = producer.send(message);
            log.info("async[]AsyncMQProducer[] send time:{}--->tag:{}--->key:{}--->msg:{}--->result {}", new Object[]{Integer.valueOf(delayTimeLevel),
                    tag, key, JSONUtil.toJsonStr(message), result});
            return result;
        } catch (Exception e) {
            return new SendResult(SendStatus.SLAVE_NOT_AVAILABLE, null, null, null, 0L);
        }
    }


    public Boolean isSendSuccess(SendResult sendResult) {
        return Boolean.valueOf(sendResult.getSendStatus().equals(SendStatus.SEND_OK));
    }
}


